use std::time::{Duration, Instant};

use actix::prelude::*;
use actix_web::{error, web};
use actix_web_actors::ws;
use serde::{Deserialize, Serialize};
use serde_json::Value;

#[derive(Debug, Serialize, Deserialize)]
struct MonitorData {
    device_data: Vec<Value>,
    mat_data: Vec<Value>,
}

/// redis 数据推送间隔
const REDIS_PUSH_INTERVAL: Duration = Duration::from_secs(3);

/// 心跳发送间隔
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);

/// 客户端超时时长
const CLIENT_TIMEOUT: Duration = Duration::from_secs(15);

pub struct MyWebSocket {
    /// 客户端至少10s发送一次心跳，否则强行断开连接
    redis: web::Data<redis::Client>,
    tag: String,
    hb: Instant,
}

impl MyWebSocket {
    pub fn new(tag: String, redis: web::Data<redis::Client>) -> Self {
        Self {
            tag,
            redis,
            hb: Instant::now(),
        }
    }

    /// 检查客户端超时时间并发送心跳
    fn hb(&self, ctx: &mut <Self as Actor>::Context) {
        ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
            if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
                println!("Websocket Client heartbeat failed, disconnecting!");
                ctx.stop();
                return;
            }

            ctx.ping(b"");
        });
    }

    fn rp(&self, ctx: &mut <Self as Actor>::Context) {
        if "visual".eq(self.tag.as_str()) {
            ctx.run_interval(REDIS_PUSH_INTERVAL, |act, ctx| {
                let mut conn = act.redis.get_connection().unwrap();

                let res: Vec<String> = redis::cmd("keys")
                    .arg("device_monitor:*")
                    .query(&mut conn)
                    .unwrap();

                let mut data_m = MonitorData {
                    device_data: vec![],
                    mat_data: vec![],
                };

                let mut data_arr = vec![];
                for s in res {
                    let mut d: Value = Value::Null;
                    let obj: Vec<String> = redis::cmd("GET").arg(s).query(&mut conn).unwrap();
                    for o in obj {
                        d = serde_json::from_str(o.as_str()).unwrap();
                        d["@type"] = Value::Null;
                    }
                    data_arr.push(d);
                }
                data_m.device_data = data_arr;
                let data_str = serde_json::to_string(&data_m).unwrap();

                ctx.text(data_str);
            });
        }
    }
}

impl Actor for MyWebSocket {
    type Context = ws::WebsocketContext<Self>;

    /// 在 actor 启动时调心跳和 redis 推送程序
    fn started(&mut self, ctx: &mut Self::Context) {
        self.hb(ctx);
        self.rp(ctx);
    }
}

/// `ws::Message` 处理器
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWebSocket {
    fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
        println!("WS: {msg:?}");
        match msg {
            Ok(ws::Message::Ping(msg)) => {
                self.hb = Instant::now();
                ctx.pong(&msg);
            }
            Ok(ws::Message::Pong(_)) => {
                self.hb = Instant::now();
            }
            Ok(ws::Message::Text(text)) => ctx.text(text),
            Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
            Ok(ws::Message::Close(reason)) => {
                ctx.close(reason);
                ctx.stop();
            }
            _ => ctx.stop(),
        }
    }
}
